package io.a.core;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

/**
 * @author weiyinglei
 * @description
 * @since 2024/7/11 11:40
 */
public final class Show2 {

    private static final Pattern SPACE = Pattern.compile(" ");

    public static void main(String[] args) throws Exception {
    if (args.length < 1) { // 保证必须有参数，此参数代表待读取文件
        System.err.println("Usage: JavaWordCount <file>");    System.exit(1);
    }

    SparkSession spark = SparkSession
            .builder() // 创建SparkSession的构建器
            .master("local[1]") // 设置部署模式
            .appName("JavaWordCount") // 设置JavaWordCount例子的应用名称
            .getOrCreate(); // 使用构建器构造SparkSession实例
    // 获取DataFrameReader，使用DataFrameReader将文本文件转换为DataFrame
    JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();
    // 使用RDD的flatMap 方法对MapPartitionsRDD进行转换  // 转换函数的作用是对每行文本进行单词拆分
    JavaRDD<String> words = lines.flatMap((FlatMapFunction<String, String>) s -> Arrays.asList(SPACE.split(s)).iterator());
    // 使用RDD的mapToPair方法对MapPartitionsRDD进行转换
        // 转换函数的作用是生成每个单词和1的对偶
        JavaPairRDD<String, Integer> ones = words.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1));
    // 使用RDD的reduceByKey方法对MapPartitionsRDD进行转换
        // 转换函数的作用是对每个单词的计数值累加
        JavaPairRDD<String, Integer> counts = ones.reduceByKey((Function2<Integer, Integer, Integer>) Integer::sum);
    // 使用RDD的collect方法对MapPartitionsRDD及其上游转换进行计算
    List<Tuple2<String, Integer>> output = counts.collect();
    for (Tuple2<?,?> tuple : output) {
        System.out.println(tuple._1() + ": " + tuple._2());
    }
    spark.stop(); // 停止SparkSession
}
}

